package com.imooc.ecommerce.kafka;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.FailureCallback;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.SuccessCallback;

import java.util.concurrent.TimeUnit;

/**
 * @description:
 * @author: sym
 * @create: 2021-11-29 20:07
 **/
@Component
@Slf4j
public class KafkaProducer {
    @Autowired
    private KafkaTemplate<String,String> kafkaTemplate;

    public void sendMessage(String key,String value,String topic){
        if (StringUtils.isBlank(value)|| StringUtils.isBlank(topic)){
            log.error("value or topic is null or empty");
            return;
        }

        ListenableFuture<SendResult<String,String>> future=StringUtils.isBlank(key)?kafkaTemplate.send(topic,value):
                kafkaTemplate.send(topic,key,value);

        //异步回调获取通知
        future.addCallback(success->{
            assert null!=success && null!=success.getRecordMetadata();
            String recordTopic=success.getRecordMetadata().topic();
            int partition = success.getRecordMetadata().partition();
            long offset = success.getRecordMetadata().offset();
            log.info("send kafka success {},{},{}",recordTopic,partition,offset);
        }, failture->{
            log.error("消费失败");
        });

        // 同步等待的方式获取通知
//        try {
////            SendResult<String, String> sendResult = future.get();
//            SendResult<String, String> sendResult = future.get(5, TimeUnit.SECONDS);
//
//            // 发送到 kafka 的 topic
//            String _topic = sendResult.getRecordMetadata().topic();
//            // 消息发送到的分区
//            int partition = sendResult.getRecordMetadata().partition();
//            // 消息在分区内的 offset
//            long offset = sendResult.getRecordMetadata().offset();
//
//            log.info("send kafka message success: [{}], [{}], [{}]",
//                    _topic, partition, offset);
//        } catch (Exception ex) {
//            log.error("send kafka message failure: [{}], [{}], [{}]",
//                    key, value, topic);
//        }
    }
}
